package com.xiaomaoguai.curator.common.zookeeper.client;

import com.xiaomaoguai.curator.common.anno.Provider;
import com.xiaomaoguai.curator.common.conf.ClientConfig;
import com.xiaomaoguai.curator.common.utils.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.reflections.Reflections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

/**
 * @ FileName: ClientSideZkThread.java
 * @ Author: WeiHui-Z
 * @ Date: 2016-11-21  14:06
 * @ Version: v1.0.0
 */
public class ClientSideZkThread implements Runnable {

	private static final Logger LOGGER = LoggerFactory.getLogger(ClientSideZkThread.class);

	private ClientConfig config;

	public ClientSideZkThread(ClientConfig config) {
		this.config = config;
	}

	@Override
	public void run() {
		String name = Thread.currentThread().getName();
		LOGGER.info("current running thread :" + name);

		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, Integer.MAX_VALUE);

		CuratorFramework curator = CuratorFrameworkFactory.builder() //
				.connectString(config.zkClusterAddress()) //  连接zookeeper的地址
				.sessionTimeoutMs(config.sessionTimeoutMs())  //session 超时时间
				.connectionTimeoutMs(config.connectionTimeoutMs()) // 连接超时时间
				.retryPolicy(retryPolicy)  //重试策略
				.build();

		final CountDownLatch latch = new CountDownLatch(1);

		curator.getConnectionStateListenable().addListener(new ConnectionStateListener() {
			@Override
			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				//连接之后，使计数减一
				if (newState == ConnectionState.CONNECTED) {
					latch.countDown();
					LOGGER.info("连接 zookeeper 成功!");
				}
			}
		});

		curator.start();
		try {
			//阻塞等待
			latch.await();
			//建立连接后，再建立路径监听器
			PathChildrenCacheListener pathChildrenCacheListener = new ClientSideZkPathChildrenCacheListener();
			// 绑定路径监听器
			// 以前是通过配置文件，这里是通过扫描package获得
			// 注意:检查问题，各种异常
			Set<String> array = fetchProcessors(config.scanPackage());
			for (String path : array) {
				// 修正
				if (!path.startsWith("/")) {
					path = "/" + path;
				}
				// 注册监听
				PathChildrenCache cache = new PathChildrenCache(curator, path, true);
				cache.getListenable().addListener(pathChildrenCacheListener);
				cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static synchronized Set<String> fetchProcessors(String[] scanPackages) throws Exception {
		Set<String> globalPathSets = new HashSet<>();
		if (null == scanPackages) {
			throw new Exception("scanPackages is null");
		}
		// 针对每个包路径
		for (String singlePackage : scanPackages) {
			// 拿到这个包路径下面的所有物理路径
			Set<String> result = fetchProcessor(singlePackage);
			for (String str : result) {
				// 逐个增加
				if (false == globalPathSets.add(str)) {// 全局包也不应该有重复的，否则必须报错
					throw new Exception("duplicated path  : " + str);
				}
			}
		}
		// 至少要有1个吧
		if (0 == globalPathSets.size()) {
			throw new Exception("globalPathSets is zero,no processor to listen on,why???");
		}
		return globalPathSets;
	}

	private static Set<String> fetchProcessor(String scanPackage) throws Exception {
		// 切换成自动扫描方式
		Set<String> pathSet = new HashSet<>();
		LOGGER.info("---------------------------------------");
		LOGGER.info("current scan package --- " + scanPackage);
		// 开始扫描
		Reflections reflections = new Reflections(scanPackage);
		Set<Class<?>> classSet = reflections.getTypesAnnotatedWith(Provider.class);
		for (Class<?> clazz : classSet) {
			// 0)获取本身类
			LOGGER.info("AnnotationClassName:" + clazz.getName());// HelloWorldService
			// 1.开始获取注解信息
			Provider provider = clazz.getAnnotation(Provider.class);
			String group = provider.group();
			String service = clazz.getName();
			String edition = provider.edition();
			// 2.最后一步
			String path = StringUtils.groupUnionServiceUnionEdition(group, service, edition) + "/PROVIDER";
			if (false == pathSet.add(path)) {
				// 注册失败,因为有重复的,单个包不容许有重复的
				throw new Exception("duplicated path: " + path);
			}
		}
		LOGGER.info("---------------------------------------");
		return pathSet;
	}
}
